package com.core.image.core.impl.bridge;

import com.core.image.core.*;
import com.core.image.utils.CircularByteBuffer;

import java.io.IOException;
import java.nio.channels.Channels;
import java.nio.channels.ReadableByteChannel;
import java.nio.channels.WritableByteChannel;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * describe:桥接调度器实现
 * 当前调度器同时实现了发送者与接受者调度逻辑
 *
 * @author 库琦
 * @date 2020/02/09/12:44
 */
public class BridgeSocketDispatcher implements ReceiveDispacher, SendDispatcher {
    //数据暂存的缓冲区
    private final CircularByteBuffer mBuffer = new CircularByteBuffer(512, true);
    // 根据缓冲区得到的读取、写入通道
    private final ReadableByteChannel readableByteChannel =  Channels.newChannel(mBuffer.getInputStream());
    private final WritableByteChannel writableByteChannel =Channels.newChannel(mBuffer.getOutputStream());
    //有数据则接受，无数据不强求填满，有多少返回多少
    private final IoArgs receiveIoArgs = new IoArgs(256,false);
    private final Receiver receiver;
    //当前是否处于发送中
    private final AtomicBoolean isSending = new AtomicBoolean();
    //用以发送IoArgs, 默认全部发送数据
    private final IoArgs sendIoArgs = new IoArgs();
    private volatile Sender sender;
    public BridgeSocketDispatcher(Receiver receiver){
        this.receiver = receiver;
    }
    public void bindSender(Sender sender){
        final Sender oldSender = this.sender;
        if (oldSender != null){
            oldSender.setSendListener(null);
        }
        synchronized (isSending){
            isSending.set(false);
        }
        mBuffer.clear();
        //设置新的发送者
        this.sender = sender;
        if (sender != null){
            sender.setSendListener(senderEventProcessor);
            requestSend();
        }
    }
    /**
     * 外部初始化好了桥接调度器后需要调用start方法开始
     */
    @Override
    public void start() {
        receiver.setReceiveListener(receiverEventProcessor);
        registerReceive();

    }
    private void registerReceive(){
        try {
            receiver.postReceiveAsync();
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    @Override
    public void stop() {

    }

    @Override
    public void send(SendPacket packet) {

    }

    @Override
    public void cancel(SendPacket packet) {

    }

    @Override
    public void sendHeartbeat() {

    }
    private void requestSend() {
        synchronized (isSending){
            final Sender sender = this.sender;
            if (isSending.get() || sender == null){
                return;
            }
            //返回True代表当前所有数据需要发送
            if (mBuffer.getAvailable() >0){
                try {
                    boolean isSucceed = sender.postSendAsync();
                    if (isSucceed){
                        isSending.set(true);
                    }

                }catch (IOException e){
                    e.printStackTrace();
                }
            }
        }
    }
    @Override
    public void close() throws IOException {

    }
    private final IoArgs.IoArgsEventProcessor receiverEventProcessor = new IoArgs.IoArgsEventProcessor() {
        @Override
        public IoArgs provideIoArgs() {
            receiveIoArgs.resetLimit();
            //一份新的IoArgs需要调用一次开始写入数据的操作
            receiveIoArgs.startWriting();
            return receiveIoArgs;
        }

        @Override
        public void onConsumeFailed(IoArgs ioArgs, Exception e) {
            e.printStackTrace();

        }

        @Override
        public void onConsumeComplete(IoArgs ioArgs) {
            ioArgs.finishWriting();
            try {
             ioArgs.writeTo(writableByteChannel);
            }catch (IOException e){
                e.printStackTrace();
            }
            registerReceive();
            // 接收数据后请求发送数据
            requestSend();


        }
    };
    private final IoArgs.IoArgsEventProcessor senderEventProcessor = new IoArgs.IoArgsEventProcessor() {
        @Override
        public IoArgs provideIoArgs() {
            try {
                int available = mBuffer.getAvailable();
                IoArgs args = BridgeSocketDispatcher.this.sendIoArgs;
                if (available > 0) {
                    args.limit(available);
                    args.startWriting();
                    args.readFrom(readableByteChannel);
                    args.finishWriting();
                    return args;
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
            return null;
        }

        @Override
        public void onConsumeFailed(IoArgs ioArgs, Exception e) {
            e.printStackTrace();
            synchronized (isSending) {
                isSending.set(false);
            }
            //继续请求发送当前的数据
            requestSend();

        }


        @Override
        public void onConsumeComplete(IoArgs ioArgs) {
            // 设置当前发送状态
            synchronized (isSending) {
                isSending.set(false);
            }
            // 继续请求发送当前的数据
            requestSend();
        }

    };
}
